package Lothar

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.{col, month, udf, year}
import org.apache.spark.sql.{DataFrame, SparkSession}

/*
* 1、初步了解每个表的数据量及用户数量。
*
* 2、由于用户收视行为信息表的记录数比较多，因此有必要对用户每次收视行为的观看时长进行基本的探索分析。
* 为了掌握用户收视行为记录中的观看时长的取值范围，以便为后续业务需求探索中用户收视行为无效数据的分析
* 探索提供帮助，统计用户收视记录观看时长的均值、最值和标准差。
*
* 3、用户收视行为信息表记录了用户每次收视节目的时长，从这些记录中可以统计每个用户平均每月的收视时长，
* 进而掌握每个用户对电视的依赖程度。
* */

object Step1 {
  private val spark: SparkSession = SparkSession.builder()
    .config("hive.metastore.uris", "thrift://master:9083")
    .config("spark.sql.warehouse.dir", "hdfs://master:8020/usr/hive/warehouse")
    .enableHiveSupport()
    .getOrCreate()

  def main(args: Array[String]): Unit = {
    // 这里的Table是经过EsToHive后写入的。直接读来用。
    var MediaIndexDF= spark.read.table("portrait.media_index")
    val BilleventDF = spark.read.table("portrait.billevent")
    val OrderDF = spark.read.table("portrait.orderData")
    val UserMsgDF = spark.read.table("portrait.usermsg")
    val UserEventDF = spark.read.table("portrait.userevent")

    // 1、初步了解每个表的数据量及用户数量。
    var answer1: Map[String, Int] = Map()
    answer1+=MediaIndexDF.getClass.getSimpleName+"数据数量"->MediaIndexDF.count()
    answer1+=BilleventDF.getClass.getSimpleName+"数据数量"->BilleventDF.count()
    answer1+=OrderDF.getClass.getSimpleName+"数据数量"->OrderDF.count()
    answer1+=UserMsgDF.getClass.getSimpleName+"数据数量"->UserMsgDF.count()
    answer1+=UserEventDF.getClass.getSimpleName+"数据数量"->UserEventDF.count()

    answer1+=MediaIndexDF.getClass.getSimpleName+"用户数量"->MediaIndexDF.select(col("phone_no")).distinct()
    answer1+=BilleventDF.getClass.getSimpleName+"用户数量"->BilleventDF.select(col("phone_no")).distinct()
    answer1+=OrderDF.getClass.getSimpleName+"用户数量"->OrderDF.select(col("phone_no")).distinct()
    answer1+=UserMsgDF.getClass.getSimpleName+"用户数量"->UserMsgDF.select(col("phone_no")).distinct()
    answer1+=UserEventDF.getClass.getSimpleName+"用户数量"->UserEventDF.select(col("phone_no")).distinct()

    // 2、 用户每次收视行为的观看时长  => media_index表duration列：单位毫秒ms    转化为分钟计算
    MediaIndexDF = MediaIndexDF.withColumn("duration",col("duration")/(1000*60))
    // 因为只说探索分析不知道究竟需要什么。神秘的甲方。直接describe看看咯，都不像是认真搞需求的。
    val MediaIndexSummary: DataFrame = MediaIndexDF.describe()
    MediaIndexDF.write.csv("./MediaIndexSummary.csv")

    // 3、每个用户平均每月的收视时长，按月分组，求平均时长咯。
    val UserTimeAvgMonth: DataFrame = MediaIndexDF.withColumn("year-month",
      year(col("origin_time")).cast(String) + month(col("origin_time")).cast(String))
      .groupBy("year-month", "phone_no")
      .avg("duration")


    /*  ========异常数据探索=========
    * 1、分析重复数据的用户，根据业务经验，这些数据表可能含有重复数据。======》各个DataFrame,distinct()一下就好了。这里就不赘述了。
    * 2、特殊线路的用户，根据客户业务人员提供的信息，owner_code的值为02、09或10的记录是特殊路线的用户，
    * 是用来测试、检验产品的，对其进行分析探索是无意义的。
    * 3、政企用户，由于该企业的客户主要来自于家庭用户，所以政企用户不纳入分析范围。根据业务人员提供的信
    * 息，政企用户的标识是owner_name的值为“EA级”、“EB级”、“EC级”、“ED级”或“EE级”。
    * */

    // 过滤异常数据 2 ==>换个表名一样用。
    val validData1: Long = MediaIndexDF.filter(col("owner_code") =!= "02" ||
                                                col("owner_code") =!= "09" ||
                                                col("owner_code") =!= "10").count()

    //  如法炮制咯。
    val validData1_1: Long = MediaIndexDF.filter(col("owner_name") =!= "EA级" ||
                                                  col("owner_name") =!= "EB级" ||
                                                  col("owner_name") =!= "EC级" ||
                                                  col("owner_name") =!= "ED级" ||
                                                  col("owner_name") =!= "EE级").count()


    /*
    * 1、分析用户收视行为无效数据。用户收视行为无效数据是指用户观看时长过短或过长的记录，引起这种现象出
    * 现的原因可能是用户频繁切换频道或者只关闭电视机而忘记关闭机顶盒。
    * ==========吐槽：为什么不说明什么是是过长或过短啊？！我又不是专业的？！！！
    * ==========10分钟以下为短，5小时以上为过长？  有标准就自己改咯。
    *
    * 2、探索消费水平标签阈值，账单信息表反映了用户每个月的消费情况，对其进行分析探索能够更好地了解用户
    * 的消费行为，制定消费水平标签的子标签及各子标签的判断阈值，从而给每个用户贴上一个合适的消费水平标签。
    *
    * 3、电视宽带入网程度标签阈值探索，用户基本信息表的open_time字段记录了用户的开户时间，利用此信息可
    * 以给用户贴上用户入网程度标签（子标签包含老用户、中等用户和新用户）。
    * */

    // 1、筛去观看过长或是过短的护具。
    MediaIndexDF.filter(col("duration") > 5 || col("duration") < (5*60))

    // 2、给每个用户贴上一个合适的消费水平标签    账单表  portrait.billevent
    val setCLabel: UserDefinedFunction = udf { (x: Double) => getConsumelable(x) }
    val billeventLabel: DataFrame = BilleventDF.groupBy("phone_no").sum("should_pay")
      .withColumn("ConsumerLabel", setCLabel(col("should_pay(sum)")))//大概估计应该可能是叫这个名字吧。

    // 3、吼吼。是和上一问同类型的替身。===》都是贴标签而已嘛。省略

  }

  // 不一定是这个区间来分段。
  def getConsumelable(num:Double): Int ={
    if(num<=10){
      return 1
    }
    else if (num<=20)
      {
        return 2
      }
    else if (num<=30)
    {
      return 3
    }
    else if (num<=40)
    {
      return 4
    }
    else
    {
      return 5
    }
  }
}
